package io.github.shenbinglife.jlock.hdfs;

import io.github.shenbinglife.jlock.LockUtils;
import io.github.shenbinglife.jlock.exception.LockCreateException;
import io.github.shenbinglife.jlock.exception.LockFailedException;
import io.github.shenbinglife.jlock.exception.UnLockFailedError;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * JHadoopFileLock
 *
 * @author shenbing
 * @date 2022/4/21
 */
public class JHadoopFileLock implements Lock {

    private static final Logger logger = LoggerFactory.getLogger(JHadoopFileLock.class);

    protected final Path lockFile;
    protected final FileSystem fs;

    public JHadoopFileLock(Path lockFile) {
        this.lockFile = lockFile;
        try {
            this.fs = lockFile.getFileSystem(new Configuration());
            if (fs.exists(lockFile) && fs.isDirectory(lockFile)) {
                throw new LockCreateException(
                        "Can not create lock file when directory exists: " + lockFile);
            }
        } catch (IOException e) {
            throw new LockCreateException("Can not open file system", e);
        }
        logger.debug("Created hadoop file lock at {}", lockFile);

    }

    @Override
    public void lock() {
        while (true) {
            if (tryLockOnce()) {
                break;
            }
            if (Thread.currentThread().isInterrupted()) {
                throw new LockFailedException("Thread interrupted when creating lock file");
            }
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        while (true) {
            if (tryLockOnce()) {
                break;
            }
            Thread.sleep(200L);
        }
    }

    @Override
    public boolean tryLock() {
        return tryLockOnce();
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        long duration = TimeUnit.MILLISECONDS.convert(time, unit);
        while (true) {
            if (tryLockOnce()) {
                return true;
            }
            Thread.sleep(200L);
            duration -= 200L;
            if (duration <= 0) {
                return false;
            }
        }
    }

    @Override
    public void unlock() {
        String cur = LockUtils.currentClientId();
        String lockedClient = fileClientId();
        if (!Objects.equals(cur, lockedClient)) {
            throw new LockFailedException("Can not unlock the file not belongs to cur client: " + cur);
        }

        try {
            if (!fs.delete(lockFile, true)) {
                throw new UnLockFailedError(
                        "Failed to delete lock file for unlock, file may not exists: "
                                + lockFile);
            }
        } catch (IOException e) {
            throw new LockFailedException("IOException when unlock file", e);
        }
    }

    protected boolean tryLockOnce() {
        boolean success = false;
        try {
            success = fs.createNewFile(lockFile);
        } catch (IOException e) {
            throw new LockFailedException("IOException when creating lock file", e);
        }
        if (success) {
            try {
                writeClientIdToFile();
            } catch (IOException e) {
                try {
                    fs.delete(lockFile, true);
                } catch (IOException e1) {
                    logger.debug("Ignore delete failed when write client id", e1);
                }
                throw new LockFailedException("Can not write client id to file", e);
            }
        }
        return success;
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException("unsupport create condition");
    }

    public boolean isLocked() {
        try {
            if (fs.exists(lockFile)) {
                if (fs.isFile(lockFile)) {
                    return true;
                } else {
                    throw new LockFailedException("Lock file is hold by directory");
                }
            }
        } catch (IOException e) {
            throw new LockFailedException("IOException when check lock file state", e);
        }
        return false;
    }


    public String fileClientId() {
        try (InputStream in = fs.open(lockFile);
             Reader inputReader = new InputStreamReader(in, StandardCharsets.UTF_8);
             BufferedReader reader = new BufferedReader(inputReader)) {
            return reader.readLine();
        } catch (IOException e) {
            throw new LockFailedException("Can not read client id from file", e);
        }
    }

    public String writeClientIdToFile() throws IOException {
        String clientId = LockUtils.currentClientId();
        try (OutputStream out = fs.create(lockFile, true);
             Writer writer = new OutputStreamWriter(out, StandardCharsets.UTF_8)) {
            writer.append(clientId);
        }
        return clientId;
    }
}
